package com.gui.connect;

import com.gui.exceptions.RedisConnectionException;
import com.gui.exceptions.RedisDataException;
import com.gui.exceptions.RedisException;
import com.gui.util.Constants;
import com.gui.util.ReplyType;
import com.gui.util.SafeEncoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.*;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

/**
 * Created by Gui on 2017/7/19.
 */
public class Processor implements Closeable {

    private static final Logger logger = LoggerFactory.getLogger(Processor.class);

    private final int DEFAULT_BUF_SIZE = 8 * 1024;

    private InputStream input;

    private byte[] buf = new byte[DEFAULT_BUF_SIZE];
    private int pos = 0;
    private int validCount = 0;     //buf的有效长度
    private int lineBegin = -1;
    private int lineLen = 0;        //不包括结束符号 /r/n

    public Processor(InputStream inputStream) {
        if (inputStream == null) {
            logger.error("inputStream should not be null");
            throw new IllegalArgumentException("inputStream should not be null");
        }
        this.input = inputStream;
    }

    public RedisReply getObjectReply() throws IOException {
        recycle();

        this.quickSkipLine(-1);
        char firstCh = (char) buf[lineBegin];
        ReplyType type = ReplyType.forValue(firstCh);
        Object value = null;

        if (type != null) {
            switch (type) {
                case STATUS:
                    value = getStatusCodeReply();
                    break;
                case INTEGER:
                    value = getIntegerReply();
                    break;
                case BULK:
                    value = getBulkReply();
                    break;
                case MULTI:
                    value = getMultiBulkReply();
                    break;
                case ERROR:
                    value = getStatusCodeReply();
                    break;
            }
        } else {
            throw new RedisDataException("unknow reply status: " + firstCh);
        }

        RedisReply reply = new RedisReply(type, value);
        logger.info("{}", reply.toString());
        return reply;

    }

    private String getStatusCodeReply() throws IOException {
        return SafeEncoder.decode(buf, lineBegin + 1, lineLen - 1);
    }

    private String getBulkReply() throws IOException {
        int len = Math.toIntExact(getIntegerReply());
        if (len != -1) {
            this.quickSkipLine(len);
            return SafeEncoder.decode(buf, lineBegin, lineLen);
        }
        return null;
    }

    private Long getIntegerReply() throws IOException {
        String reply = SafeEncoder.decode(buf, lineBegin + 1, lineLen - 1);
        return Long.parseLong(reply);
    }

    private List<String> getMultiBulkReply() throws IOException {
        int num = Math.toIntExact(getIntegerReply());
        if (num != -1) {
            List<String> list = new ArrayList<>();
            for (int i = 0; i < num; i ++) {
                this.quickSkipLine(-1);
                int len = Math.toIntExact(getIntegerReply());
                this.quickSkipLine(len);
                String reply = SafeEncoder.decode(buf, lineBegin, lineLen);
                list.add(reply);
            }
            return  list;
        }
        return null;
    }


    //------------------------------------------------ migrate begin ---------------------------------------//

    public void saveRDBData(OutputStream outputStream) throws IOException {
        this.quickSkipLine(-1);
        long totalLen = getIntegerReply();
        if (totalLen <= 0) {
            logger.error("rdb length error: {}", totalLen);
            throw new RedisException("error rdb length: " + totalLen);
        }

        if (pos < validCount) {
            outputStream.write(buf, pos, validCount - pos);
            totalLen -= (validCount - pos);
        }

        recycle();

        int len;
        while (totalLen > 0 && (len = input.read(buf, 0, (int) Math.min(totalLen, (long) buf.length))) != -1) {
            totalLen -= len;
            outputStream.write(buf, 0, len);
        }
    }

    public byte[] loadSingleAOF() throws IOException {
        int from, to;

        recycle();
        this.quickSkipLine(-1);
        from = lineBegin;

        if (buf[lineBegin] != ReplyType.MULTI.value) {
            throw new RedisDataException("parsed aof error");
        }
        long count = getIntegerReply();
        for (int i = 0; i < count; i ++) {
            this.quickSkipLine(-1);
            long len = getIntegerReply();
            this.quickSkipLine((int) len);
        }
        to = lineBegin + lineLen + 2;
        return Arrays.copyOfRange(buf, from, to);
    }


    //----------------------------------------------migrate end ------------------------------------//


    /**
     * 根据Redis的通信协议可知，Redis发送的命令或数据一律以 \r\n 结束，所以我们需要找到一行命令或数据的开始和结束位置
     * 但是，对于某些数据，协议中提供了其字节长度，所以这里必须将其长度作为参数传入；否则，传入-1，我们将以为 \r\n
     * 作为结束符
     * @param len
     * @throws IOException
     */
    private void quickSkipLine(int len) throws IOException {
        byte chr = 0;
        while (len < 0) {
            //剔除无效数据， 例如   \r\nabc\r\n
            if (pos >= validCount) {
                if (!fill(-1)) {
                    throw new IOException("Unable to get more data.");
                }
            }
            chr = buf[pos ++];
            if (!(chr == Constants.CR || chr == Constants.LF)) {
                pos --;
                break;
            }
        }

        lineBegin = pos;
        while (true) {
            if (pos >= validCount) {
                if (!fill(lineBegin +len)) {
                    throw new IOException("Unable to get more data.");
                }
            }
            chr = buf[pos ++];
            if (chr == Constants.CR) {
                lineLen = pos - lineBegin - 1;
            } else if (chr == Constants.LF) {
                if (len < 0 || lineLen >= len) {
                    break;
                }
            }
        }

    }

    /**
     * 获取字节
     * @param len 如果已知将要获取的字节长度，可将字节长度传入
     * @return
     */
    private boolean fill(int len) throws IOException {
        if (buf.length - validCount < 2000) {
            //这种情况下，可认为需要接受字节非常多
            buf = Arrays.copyOf(buf, Math.max(buf.length + (buf.length >> 1), len));
        }
        int nRead = input.read(buf, validCount, buf.length - validCount);
        if (nRead > 0) {
            validCount += nRead;
        }
        return nRead > 0;
    }

    private void recycle() {
        //如果buf中还有数据未处理，则不进行循环利用
        //为了防止防止buf累计过大，当buf的长度为default_buf_size的5次方时，强制循环利用

        int remainCount = validCount - pos;     //剩余数据量
        boolean oversize = buf.length > DEFAULT_BUF_SIZE * 8;       //是否过大
        boolean resizeBuf = oversize || ((remainCount <= 0) && (buf.length > DEFAULT_BUF_SIZE * 2.5));

        if (resizeBuf) {
            int newSize = Math.max(remainCount, DEFAULT_BUF_SIZE);
            byte[] result = new byte[newSize];
            if (remainCount > 0) {
                System.arraycopy(buf, pos, result, 0, remainCount);
            }
            buf = result;

            pos = 0;
            validCount = remainCount;
        } else if (remainCount <= 0){
            pos = 0;
            validCount = 0;
        }

    }

    public void close() throws IOException {
        if (input != null) {
            input.close();
        }
    }
}
